package com.atguigu.day04;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import com.atguigu.util.UserViewCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @description :
 * @autor :lyh
 */
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.username)
                .process(new MyTumblingProcessTimeWindow(10 * 1000L))
                .print();

        env.execute();

    }

    public static class MyTumblingProcessTimeWindow extends KeyedProcessFunction<String, ClickEvent, UserViewCountPerWindow> {
        private long windowSize;
        // key: 窗口信息
        // value: 属于窗口的累计器
        private MapState<Tuple2<Long, Long>, Long> mapState;

        public MyTumblingProcessTimeWindow(long windowSize) {
            this.windowSize = windowSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Tuple2<Long, Long>, Long>(
                            "window-accumulator",
                            Types.TUPLE(Types.LONG, Types.LONG),
                            Types.LONG
                    )
            );
        }

        @Override
        public void processElement(ClickEvent value, KeyedProcessFunction<String, ClickEvent, UserViewCountPerWindow>.Context ctx, Collector<UserViewCountPerWindow> out) throws Exception {
            long currentTime = ctx.timerService().currentProcessingTime();
            long windowStartTime = currentTime - currentTime % windowSize;
            long windowEndTime = windowStartTime + windowSize;

            Tuple2<Long, Long> windowInfo = Tuple2.of(windowStartTime, windowEndTime);

            if (!mapState.contains(windowInfo)) {
                mapState.put(windowInfo, 1L);
            } else {
                mapState.put(windowInfo, mapState.get(windowInfo) + 1L);
            }
            ctx.timerService().registerProcessingTimeTimer(windowEndTime - 1L);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, ClickEvent, UserViewCountPerWindow>.OnTimerContext ctx, Collector<UserViewCountPerWindow> out) throws Exception {
            long windowEnd = timestamp + 1L;
            long windowStart = windowEnd - windowSize;
            Tuple2<Long, Long> windowInfo = Tuple2.of(windowStart, windowEnd);

            Long count = mapState.get(windowInfo);

            out.collect(new UserViewCountPerWindow(
                    ctx.getCurrentKey(),
                    count,
                    windowStart,
                    windowEnd
            ));
            mapState.remove(windowInfo);
        }
    }


}
